package com.IntelligenceInsight.Myth.service;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class HDFSService {

    private FileSystem hdfs = null;
    private Set<String> filesSet = null;

    private volatile static HDFSService uniqueInstance;

    public static HDFSService getInstance() {
        if(uniqueInstance == null) {
            synchronized (HDFSService.class) {
                if(uniqueInstance == null) {
                    uniqueInstance = new HDFSService();
                }
            }
        }
        return uniqueInstance;
    }

    public HDFSService() {
       // System.setProperty("hadoop.home.dir", "/");
        System.setProperty("hadoop.home.dir", "D:\\Programsworkspace\\hadoop-3.2.0");
        Configuration configuration = new Configuration();

        String hdfsUriString = "hdfs://192.168.1.121:9000";
        String hdfsUserName = "hadoop";

        URI hdfsUri = null;
        try {
            hdfsUri = new URI(hdfsUriString);
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }

        try {
            hdfs = FileSystem.get(hdfsUri, configuration, hdfsUserName);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    /*
     * 创建文件夹
     */
    private boolean createDir(String dirPath){
        Path path = new Path(dirPath);
        try {
            return hdfs.mkdirs(path);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /*
     * 删除文件夹
     */
    public boolean removeDir(String dirPath){
        Path path = new Path(dirPath);
        try {
            return hdfs.delete(path,true);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /*
     * 重命名文件夹
     */
    public boolean renameDir(String oldDirPath, String newDirPath){
        Path oldPath = new Path(oldDirPath);
        Path newPath = new Path(newDirPath);

        try {
            return hdfs.rename(oldPath,newPath);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /*
     * 遍历文件夹
     */
    public Set<String> listDir(String dirPath){
        Path path = new Path(dirPath);

        FileStatus[] files = null;
        Set<String> set = new HashSet<>();

        try {
            files = hdfs.listStatus(path);

            if(files.length == 0) {
                return null;
            }
            for (FileStatus f : files) {
                set.add(f.getPath().toUri().getPath());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return set;
    }

    /*
     * 递归遍历文件夹
     */
    public Set<String> recursiveDir(String dirPath){
        if(filesSet == null) {
            filesSet = new HashSet<>();
        } else {
            filesSet.clear();
        }

        _recursiveDir(dirPath);

        return filesSet;
    }

    private void _recursiveDir(String dirPath) {
        Path path = new Path(dirPath);

        FileStatus[] files = null;

        try {
            files = hdfs.listStatus(path);

            if(files.length == 0) {
                return;
            }

            for (FileStatus f : files) {
                if (f.isFile()) {
                    filesSet.add(f.getPath().toUri().getPath());
                } else {
                    _recursiveDir(f.getPath().toUri().getPath());
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /*
     * 路径是否存在
     */
    public boolean isExists(String path) {
        try {
            return hdfs.exists(new Path(path));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /*
     * 是否是文件
     */
    public boolean isFile(String path) {
        try {
            return hdfs.isFile(new Path(path));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /*
     * 是否是文件夹
     */
    public boolean isDirectory(String path) {
        try {
            return hdfs.isDirectory(new Path(path));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /*
     * 从HDFS下载文件到本地
     */
    public void getFileFromHDFS(String hdfsPath, String localPath){
        Path _hdfsPath = new Path(hdfsPath);
        Path _localPath = new Path(localPath);

        try {
            hdfs.copyToLocalFile(_hdfsPath,_localPath);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /*
     * 上传本地文件至HDFS
     */
    public boolean putFileToHDFS(String localPath, String hdfsPath) {
        Path _localPath = new Path(localPath);
        Path _hdfsPath = new Path(hdfsPath);

        try {
//            if(isExists(hdfsPath)) {
//                System.out.println("HDFS path is exists.");
//                return false;
//            }
            hdfs.copyFromLocalFile(_localPath, _hdfsPath);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return true;
    }

    /*
     * 复制HDFS文件
     */
    public void copyHDFSFile(String from, String to){
        Path fromPath = new Path(from);
        Path toPath = new Path(to);

        FSDataInputStream fsDataInputStream = null;
        FSDataOutputStream fsDataOutputStream = null;

        try {
            fsDataInputStream = hdfs.open(fromPath);
            fsDataOutputStream = hdfs.create(toPath);

            IOUtils.copyBytes(fsDataInputStream, fsDataOutputStream,4096,false);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if(fsDataInputStream != null) {
                IOUtils.closeStream(fsDataInputStream);
            }
            if(fsDataOutputStream != null) {
                IOUtils.closeStream(fsDataOutputStream);
            }
        }
    }

    /*
     * 读取HDFS文件内容
     */
    public List<String> readHDFSFile(String hdfsPath) {
        Path path = new Path(hdfsPath);

        FSDataInputStream fsDataInputStream = null;
        BufferedReader bufferedReader = null;
        List<String> output = new ArrayList<>();

        try {
            fsDataInputStream = hdfs.open(path);
            bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream));

            String line = null;
            while((line = bufferedReader.readLine()) != null) {
                output.add(line);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return output;
    }

//    public void writeFileToHDFS(String localPath, String hdfsPath) {
//        Path path = new Path(hdfsPath);
//
//        FileInputStream fileInputStream = null;
//        FSDataOutputStream fsDataOutputStream = null;
//
//        try {
//            fileInputStream = new FileInputStream(new File(localPath));
//            fsDataOutputStream = hdfs.create(path);
//
//            IOUtils.copyBytes(fileInputStream, fsDataOutputStream, 4096, false);
//        } catch (IOException e) {
//            e.printStackTrace();
//        } finally {
//            if(fileInputStream != null) {
//                IOUtils.closeStream(fileInputStream);
//            }
//            if(fsDataOutputStream != null) {
//                IOUtils.closeStream(fsDataOutputStream);
//            }
//        }
//    }

    /*
     * 将数据流写入HDFS
     */
    public void writeStreamToHDFS(InputStream inputStream, String hdfsPath) {
        FSDataOutputStream fsDataOutputStream = null;
        Path path = new Path(hdfsPath);

        try {
            fsDataOutputStream = hdfs.create(path);
            IOUtils.copyBytes(inputStream, fsDataOutputStream, 4096, false);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if(inputStream != null) {
                IOUtils.closeStream(inputStream);
            }
            if(fsDataOutputStream != null) {
                IOUtils.closeStream(fsDataOutputStream);
            }
        }
    }
}